#ifndef __M_CONSUMER_H__
#define __M_CONSUMER_H__

#include "../mqcommon/mq_logger.hpp"
#include "../mqcommon/mq_helper.hpp"
#include "../mqcommon/mq_msg.pb.h"
#include <iostream>
#include <unordered_map>
#include <mutex>
#include <memory>
#include <vector>

namespace lywmq
{
using ConsumerCallback = std::function<void(const std::string, const BasicProperties* bp, const std::string)>;
// 消费者
struct Consumer
{
    using ptr = std::shared_ptr<Consumer>;

    std::string tag; // 消费者标识
    std::string qname; // 消费者订阅队列名称
    bool auto_ack; // 自动确认标志
    ConsumerCallback callback;

    Consumer()
    {
        DLOG("new Consumer: %p", this);
    }
    Consumer(const std::string& ctag, const std::string& queue_name, bool ack_flag, const ConsumerCallback& cb)
    :tag(ctag), qname(queue_name), auto_ack(ack_flag), callback(cb)
    {
        DLOG("new Consumer: %p", this);
    }
    ~Consumer() 
    {
        DLOG("del Consumer: %p", this);
    }
};

// 队列
class QueueConsumer
{
public:
    using ptr = std::shared_ptr<QueueConsumer>;
    QueueConsumer(const std::string& qname)
    :_qname(qname), _rr_seq(0)
    {}
    // 新增消费者
    Consumer::ptr create(const std::string& ctag, const std::string& queue_name, bool ack_flag, const ConsumerCallback& cb)
    {
        std::unique_lock<std::mutex> lock(_mutex);
        // 判断重复
        for (auto& consumer: _consumers)
        {
            if (consumer->tag == ctag)
                return Consumer::ptr();
        }
        // 新增
        auto consumer = std::make_shared<Consumer>(ctag, queue_name, ack_flag, cb);
        _consumers.push_back(consumer);
        return consumer;
    }
    // 移除消费者
    void remove(const std::string& ctag)
    {
        std::unique_lock<std::mutex> lock(_mutex);
        for (auto it = _consumers.begin(); it != _consumers.end(); it++)
        {
            if ((*it)->tag == ctag)
            {
                _consumers.erase(it);
                return;
            }
        }
    }
    // 获取消费者（RR轮转）
    Consumer::ptr choose()
    {
        if (_consumers.size() == 0)
            return Consumer::ptr();

        std::unique_lock<std::mutex> lock(_mutex);
        int idx = _rr_seq % _consumers.size();
        _rr_seq++;
        return _consumers[idx];
    }
    bool empty()
    {
        std::unique_lock<std::mutex> lock(_mutex);
        return _consumers.size() == 0;
    }
    bool exists(const std::string& ctag)
    {
        std::unique_lock<std::mutex> lock(_mutex);
        for (auto& consumer: _consumers)
        {
            if (consumer->tag == ctag)
                return true;
        }
        return false;
    }
    void clear()
    {
        std::unique_lock<std::mutex> lock(_mutex);
        _consumers.clear();
        _rr_seq = 0;
    }
private:
    std::string _qname;
    std::mutex _mutex;
    uint64_t _rr_seq; // 轮转序号
    std::vector<Consumer::ptr> _consumers;
};

// 管理消费者队列
class ConsumerManager
{
public:
    using ptr = std::shared_ptr<ConsumerManager>;
    ConsumerManager(){}
    
    void initQueueConsumer(const std::string& qname)
    {
        std::unique_lock<std::mutex> lock(_mutex);
        // 判断重复
        auto it = _qconsumers.find(qname);
        if (it != _qconsumers.end()) return;
        // 新增
        QueueConsumer::ptr qconsumer = std::make_shared<QueueConsumer>(qname);
        _qconsumers.insert(std::make_pair(qname, qconsumer));
    }
    void destroyQueueConsumer(const std::string& qname)
    {
        std::unique_lock<std::mutex> lock(_mutex);
        _qconsumers.erase(qname);
    }
    Consumer::ptr create(const std::string& ctag, const std::string& queue_name, bool ack_flag, const ConsumerCallback& cb)
    {
        std::unique_lock<std::mutex> lock(_mutex);
        QueueConsumer::ptr qcp;
        {
            // 判断重复
            auto it = _qconsumers.find(queue_name);
            if (it == _qconsumers.end())
            {
                DLOG("没有找到队列的 %s 的消费者管理句柄！", queue_name.c_str());
                return Consumer::ptr();
            }
            qcp = it->second;
        }
        return qcp->create(ctag, queue_name, ack_flag, cb);
    }
    void remove(const std::string& ctag, const std::string& queue_name)
    {
        std::unique_lock<std::mutex> lock(_mutex);
        QueueConsumer::ptr qcp;
        {
            // 判断重复
            auto it = _qconsumers.find(queue_name);
            if (it == _qconsumers.end())
            {
                DLOG("没有找到队列的 %s 的消费者管理句柄！", queue_name.c_str());
                return;
            }
            qcp = it->second;
        }
        return qcp->remove(ctag);
    }
    Consumer::ptr choose(const std::string& queue_name)
    {
        std::unique_lock<std::mutex> lock(_mutex);
        QueueConsumer::ptr qcp;
        {
            // 判断重复
            auto it = _qconsumers.find(queue_name);
            if (it == _qconsumers.end())
            {
                DLOG("没有找到队列的 %s 的消费者管理句柄！", queue_name.c_str());
                return Consumer::ptr();
            }
            qcp = it->second;
        }
        return qcp->choose();
    }
    bool empty(const std::string& queue_name)
    {
        std::unique_lock<std::mutex> lock(_mutex);
        QueueConsumer::ptr qcp;
        {
            // 判断重复
            auto it = _qconsumers.find(queue_name);
            if (it == _qconsumers.end())
            {
                DLOG("没有找到队列的 %s 的消费者管理句柄！", queue_name.c_str());
                return false;
            }
            qcp = it->second;
        }
        return qcp->empty();
    }
    bool exists(const std::string& ctag, const std::string& queue_name)
    {
        std::unique_lock<std::mutex> lock(_mutex);
        QueueConsumer::ptr qcp;
        {
            // 判断重复
            auto it = _qconsumers.find(queue_name);
            if (it == _qconsumers.end())
            {
                DLOG("没有找到队列的 %s 的消费者管理句柄！", queue_name.c_str());
                return false;
            }
            qcp = it->second;
        }
        return qcp->exists(ctag);
    }
    void clear()
    {
        std::unique_lock<std::mutex> lock(_mutex);
        _qconsumers.clear();
    }
private:
    std::mutex _mutex;
    std::unordered_map<std::string, QueueConsumer::ptr> _qconsumers;
};

}

#endif